package com.tenthoundsqps.service.sync;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.tenthoundsqps.entity.Order;
import com.tenthoundsqps.service.sync.dto.OrderSyncMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

import java.util.UUID;

/**
 * 基于Kafka的订单异步同步服务
 * 通过消息队列实现订单数据的异步同步到Elasticsearch
 * 
 * 异步同步优势：
 * 1. 解耦：订单服务与ES同步服务解耦
 * 2. 异步：避免同步操作影响订单处理性能
 * 3. 可靠：通过消息队列保证数据不丢失
 * 4. 扩展：支持多个消费者并行处理
 */
@Service
public class OrderKafkaSyncService {
    
    private static final Logger logger = LoggerFactory.getLogger(OrderKafkaSyncService.class);
    
    /**
     * JSON对象映射器
     * 用于消息的序列化和反序列化
     */
    private final ObjectMapper objectMapper = new ObjectMapper();
    
    /**
     * Kafka消息模板
     * 用于向Kafka发送订单同步消息
     */
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;
    
    // Kafka主题名称
    private static final String ORDER_SYNC_TOPIC = "order-sync-topic";
    private static final String ORDER_DELETE_TOPIC = "order-delete-topic";
    
    /**
     * 发送订单同步消息到Kafka
     * 用于订单创建或更新时通知ES同步服务
     * 
     * @param order 订单对象
     */
    public void sendOrderSyncMessage(Order order) {
        try {
            // 构造消息对象
            OrderSyncMessage message = new OrderSyncMessage(order.getId(), "SYNC");
            
            // 将消息对象序列化为JSON字符串
            String messageJson = objectMapper.writeValueAsString(message);
            
            // 发送消息到Kafka
            kafkaTemplate.send(ORDER_SYNC_TOPIC, String.valueOf(order.getId()), messageJson);
            
            logger.info("Order sync message sent to Kafka, orderId: {}", order.getId());
        } catch (Exception e) {
            logger.error("Failed to send order sync message to Kafka, orderId: {}", order.getId(), e);
        }
    }
    
    /**
     * 发送订单删除消息到Kafka
     * 用于订单删除时通知ES同步服务
     * 
     * @param orderId 订单ID
     */
    public void sendOrderDeleteMessage(Long orderId) {
        try {
            // 构造消息对象
            OrderSyncMessage message = new OrderSyncMessage(orderId, "DELETE");
            
            // 将消息对象序列化为JSON字符串
            String messageJson = objectMapper.writeValueAsString(message);
            
            // 发送消息到Kafka
            kafkaTemplate.send(ORDER_DELETE_TOPIC, String.valueOf(orderId), messageJson);
            
            logger.info("Order delete message sent to Kafka, orderId: {}", orderId);
        } catch (Exception e) {
            logger.error("Failed to send order delete message to Kafka, orderId: {}", orderId, e);
        }
    }
}